#include "libcomm.h"

MessageQueue::MessageQueue(size_t maxCapacity)
	:m_maxCapacity(maxCapacity)
{
}

MessageQueue::~MessageQueue(void)
{
	Clear();
}

bool MessageQueue::Push(const void* ptr, size_t size, bool atFront, unsigned int timeoutMs)
{
	unsigned int intervalTime = 10;
	unsigned int maxTimes = timeoutMs / intervalTime;
	unsigned int i = 0;
	while(true)
	{
		do 
		{
			ThreadLockGuard guard(this);
			if(-1 != m_maxCapacity)
			{
				if(this->Capacity() + size > m_maxCapacity)
				{
					if(i == 0)
					{
						LOGWARN("%s: buffer exceed max capatity(%ubytes), waiting(%ums) to reduce...",__FUNCTION__,m_maxCapacity,timeoutMs);
					}
					break;
				}
			}
			
			Buffer* buffer = new Buffer(ptr,size);
			MYLIB_ASSERT_NOTNULL(buffer);
			m_bufferList.Append(buffer,atFront);
			if(i != 0)
			{
				LOGWARN("%s: operation success after waiting(%ums).",__FUNCTION__, i*intervalTime);
			}
			m_sem.SetSignal();
			return true;

		} while (false);

		if(i >= maxTimes)
		{
			LOGWARN("%s: operation failure, waiting(%ums) timeout.",__FUNCTION__, i*intervalTime);
			break;
		}
		++i;
		Thread::Sleep(intervalTime);
	}
	
	return false;
}

bool MessageQueue::Push(const Buffer& buffer, bool atFront, unsigned int timeoutMs)
{
	return Push(buffer.get(),buffer.size(),atFront,timeoutMs);
}

Buffer* MessageQueue::Pop(unsigned int timeoutMs,  bool atFront)
{
	Buffer* buffer = NULL;
	if(m_sem.WaitSignal(timeoutMs))
	{
		ThreadLockGuard guard(this);
		m_bufferList.Pop(buffer,atFront);
	}
	return buffer;
}

Buffer* MessageQueue::Peek(bool atFront)
{
	ThreadLockGuard guard(this);
	Buffer* buffer = NULL;
	m_bufferList.Peek(buffer,atFront);
	return buffer;
}

size_t MessageQueue::Capacity()
{
	ThreadLockGuard guard(this);
	size_t total = 0;
	foreach(Buffer*,item,m_bufferList)
	{
		total += item->size();
	}
	return total;
}

void MessageQueue::Flat(Buffer& buffer)
{
	ThreadLockGuard guard(this);
	size_t bytes = 0;
	char* ptr = NULL;
	buffer.resize(this->Capacity());
	ptr = buffer.get();
	foreach(Buffer*,item,m_bufferList)
	{
		MYLIB_ASSERT(bytes + item->size() <= buffer.size());
		memcpy(ptr + bytes,item->get(),item->size());
		bytes += item->size();
	}
	MYLIB_ASSERT(bytes==buffer.size());
}

// notice to reduce semaphore count.
void MessageQueue::Clear()
{
	ThreadLockGuard guard(this);
	do 
	{
		Buffer* buffer = Pop(0);
		if(NULL == buffer)
			break;
		delete buffer;
	} while (true);
	MYLIB_ASSERT(0 == m_bufferList.Count());
}

